package com.storm.queue.thread

import com.storm.queue.NextTimeHolder
import com.storm.queue.condition
import com.storm.queue.core.DelayOperation
import com.storm.queue.lock
import org.slf4j.LoggerFactory
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

object MoveJobThread {

    private val logger = LoggerFactory.getLogger(MoveJobThread::class.java)

    @Volatile
    private var toStop = false

    /**
     * 搬运线程:  将 bucket中的延迟有序队列zSet尝试move到待消费队列List
     */
    private val BUCKET_MOVE_TO_LIST = Executors.newSingleThreadExecutor()

    /**
     * 搬运线程
     */
    fun doMoveJob(delayJobOperation: DelayOperation) {
        BUCKET_MOVE_TO_LIST.execute {
            while (!toStop) {
                // 睡一秒钟:  避免QPS过高,并且频繁的有更小的的时间戳导致执行
                // 搬运操作频率过高没有必要(而且一次搬运是搬运多个元素的,一秒钟之内更新最小的时间戳就行了)
                // 策略是: 最慢每分钟执行一次; 最快一秒一次;
                logger.info("开始进行搬运工作")
                try {
                    Thread.sleep(1000)
                    var timeout: Long
                    if ((NextTimeHolder.nextTime.get() - System.currentTimeMillis()).also { timeout = it } > 0) {
                        lock.lock()
                        try {
                            logger.info("进入睡眠")
                            condition.await(timeout, TimeUnit.MILLISECONDS)
                        } catch (e: InterruptedException) {
                            logger.warn("runMove2ReadyThread wait被打断")
                            e.printStackTrace()
                        }finally {
                            lock.unlock()
                        }
                    } else {
                        logger.info("尝试搬运一次")
                        val score: Long = delayJobOperation.moveToQueue()
                        //重新设置nextTime值;这里不调用tryUpdate来更新;而是直接set;以这里的为准
                        NextTimeHolder.nextTime.set(score)
                    }
                } catch (e: InterruptedException) {
                    //停机打断睡眠
                } catch (ec: Exception) {
                    logger.error(ec.message, ec)
                    //redis 网络挂掉  睡5s
                    try {
                        Thread.sleep(5000)
                    } catch (e: InterruptedException) {
                        e.printStackTrace()
                    }
                }
            }
        }
    }

    fun toStop() {
        logger.info("搬运线程关闭.....")
        toStop = true
        BUCKET_MOVE_TO_LIST.shutdown()
    }
}